package com.cgs.query.remote.netty

import com.cgs.query.exception.QueryException
import com.cgs.query.handler.HELLO_METHOD
import com.cgs.query.handler.Request
import com.cgs.query.handler.Response
import com.cgs.query.handler.SyncHandler
import com.cgs.query.handler.SyncHandler.Companion.PROJECT_SYNC_METHOD_NAME
import com.cgs.query.service.IQueryService
import com.cgs.query.util.CastUtils
import io.netty.bootstrap.ServerBootstrap
import io.netty.buffer.ByteBuf
import io.netty.channel.*
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioServerSocketChannel
import io.netty.handler.codec.ByteToMessageDecoder
import io.netty.handler.codec.MessageToByteEncoder
import io.netty.handler.logging.LogLevel
import io.netty.handler.logging.LoggingHandler
import mu.KotlinLogging
import java.net.InetSocketAddress
import io.netty.util.concurrent.GlobalEventExecutor
import io.netty.channel.group.DefaultChannelGroup
import io.netty.util.internal.PlatformDependent
import java.lang.reflect.Method
import java.net.SocketAddress
import java.util.*
import java.util.concurrent.ConcurrentMap


private val logger = KotlinLogging.logger { }


class NettyServer(
        private val queryService: IQueryService,
        private val localAddress: InetSocketAddress,
        private val serialization: Serialization,
        private var syncHandler: SyncHandler? = null
) {
    companion object {
        val methods: Array<out Method> = IQueryService::class.java.methods!!
    }

    val channelGroup = DefaultChannelGroup(GlobalEventExecutor.INSTANCE)
    val channelActiveLogQueue = ArrayDeque<NettyServerLog>()
    val channelId2AppName: ConcurrentMap<ChannelId, String> = PlatformDependent.newConcurrentHashMap<ChannelId, String>()

    private var boss: EventLoopGroup? = null
    private var worker: EventLoopGroup? = null
    private var channel: Channel? = null

    inner class NettyDecoder(private val clzz: Class<*>) : ByteToMessageDecoder() {
        @Throws(Exception::class)
        override fun decode(ctx: ChannelHandlerContext, byteBuf: ByteBuf, out: MutableList<Any>) {
            if (byteBuf.readableBytes() < 4) {
                return
            }
            byteBuf.markReaderIndex()
            val dataLength = byteBuf.readInt()
            if (dataLength < 0) {
                ctx.close()
            }
            if (byteBuf.readableBytes() < dataLength) {
                byteBuf.resetReaderIndex()
                return  // fix 1024k buffer splice limix
            }
            val data = ByteArray(dataLength)
            byteBuf.readBytes(data)

            val obj = serialization.deserialize(data, clzz)
            out.add(obj)
        }

    }

    inner class NettyEncoder : MessageToByteEncoder<Response>() {

        @Throws(Exception::class)
        override fun encode(ctx: ChannelHandlerContext, msg: Response, out: ByteBuf) {
            val data = serialization.serialize(msg)
            out.writeInt(data.size)
            out.writeBytes(data)
        }
    }

    @ChannelHandler.Sharable
    inner class NettyServerHandler : ChannelInboundHandlerAdapter() {

        private val pong = Response(0L, "pong")

        @Throws(Exception::class)
        override fun channelActive(ctx: ChannelHandlerContext) {
            val channel = ctx.channel()
            channelGroup.add(channel)
            channelActiveLogQueue.addFirst(NettyServerLog("up", channel?.remoteAddress()))
        }

        override fun channelInactive(ctx: ChannelHandlerContext?) {
            val channel = ctx?.channel()
            channelGroup.remove(channel)
            channelActiveLogQueue.addFirst(NettyServerLog("down", channel?.remoteAddress()))
        }

        @Throws(Exception::class)
        override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
            var response: Response? = null
            if (msg is Request) {
                try {
                    response = when (msg.methodName) {
                        "ping" -> pong
                        PROJECT_SYNC_METHOD_NAME -> {
                            if (msg.args.isNotEmpty()) {
                                val result = syncHandler?.originServerResponse(msg.args[0] as? String
                                        ?: throw QueryException("同步失败:无效的参数[${msg.args[0]}]"))
                                Response(msg.requestId, result)
                            } else {
                                throw QueryException("不支持的方法:[${msg.methodName}]")
                            }
                        }
                        HELLO_METHOD -> {
                            channelId2AppName[ctx.channel().id()] = CastUtils.any2String(msg.args[0],"unknown")
                            Response(msg.requestId, HELLO_METHOD)
                        }
                        else -> {
                            val service = queryService
                            val method = methods.firstOrNull { it.name == msg.methodName }
                                    ?: throw QueryException("不支持的方法:[${msg.methodName}]")
                            val argArray = msg.args.toTypedArray()
                            val result = method.invoke(service, *argArray)
                            Response(msg.requestId, result)
                        }
                    }
                } catch (e: Exception) {
                    logger.error { e }
                    response = Response(msg.requestId, null).apply { exception = e }
                }
                ctx.writeAndFlush(response)
            } else {
                logger.debug("不支持的数据类型:", msg.javaClass)
                ctx.writeAndFlush(response)
            }
        }
    }

    @Throws(RuntimeException::class)
    fun start() {
        boss = NioEventLoopGroup(1)
        worker = NioEventLoopGroup()
        val initializer = object : ChannelInitializer<SocketChannel>() {
            @Throws(Exception::class)
            override fun initChannel(ch: SocketChannel) {
                ch.pipeline().addLast(NettyDecoder(Request::class.java)).addLast(NettyEncoder())
                        .addLast(NettyServerHandler())
            }
        }
        val future = ServerBootstrap().group(boss, worker).localAddress(localAddress)
                .handler(LoggingHandler(LogLevel.INFO)).childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_KEEPALIVE, true).channel(NioServerSocketChannel::class.java)
                .childHandler(initializer).bind()
        try {
            channel = future.sync().addListener { f ->
                if (f.isSuccess) {
                    logger.info("服务启动成功 : 地址 $localAddress")
                }
            }.channel()
        } catch (e: InterruptedException) {
            stop()
            throw QueryException("Netty服务启动失败", e)
        }

    }

    fun stop() {
        boss!!.shutdownGracefully()
        worker!!.shutdownGracefully()
        channelGroup.close()
        if (channel != null && channel!!.isOpen) {
            channel!!.close()
        }
    }
}

/**
 * 日志信息
 */
data class NettyServerLog(
        /**类型*/
        val type: String,
        /**地址*/
        val socketAddress: SocketAddress?) {

    val time = System.currentTimeMillis()
}
